/**
 * Created by zhangjinxia on 2017/10/12.
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import jedis.Redis;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.client.transport.TransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import util.ElasticsearchUtil;
import util.PropertyUtil;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

public class CanalClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(CanalClient.class);
    private static Logger errorLogger = LoggerFactory.getLogger("canalClientErrorLogger");
    private static final Logger ES_ERROR = LoggerFactory.getLogger("esErrorLogger");

    //我接的零工
    private static final String ORDER_USERID_STATUS099_LIST = "order-userid-status099-list";
    private static final String ORDER_USERID_STATUS020_LIST = "order-userid-status020-list";
    private static final String ORDER_USERID_STATUS030_LIST = "order-userid-status030-list";
    private static final String ORDER_USERID_STATUS040_LIST = "order-userid-status040-list";

    //我发的零工
    private static final String JOB_USERID_STATUS099_LIST = "job-userid-status099-list";
    private static final String JOB_USERID_STATUS020_LIST = "job-userid-status020-list";
    private static final String JOB_USERID_STATUS030_LIST = "job-userid-status030-list";
    private static final String JOB_USERID_STATUS040_LIST = "job-userid-status040-list";

    //订单状态(6.有人雇佣 7.拒绝雇佣 8等待雇员同意 9雇员拒绝,10.等待同意,11.拒绝接单,
    //20.等待开工,21.正在干活中,22.等待验收,30.待收款,40.已完成,50.已取消,60.强行终止)',
    //任务的订单列表
    private static final String JOBORDER_JOBID_STATUS099_LIST = "joborder-jobid-status099-list";
    private static final String JOBORDER_STATUS_ALL = PropertyUtil.getValue("joborder_status_all");

    //订单信息
    private static final String ORDER_ORDERID_MAP = "order-orderid-map";
    private static final String ORDER_ORDERID_CANCEL_MAP = "order-orderid-cancel-map";
    private static final String EVALUATE_EE_ORDERID_MAP = "evaluate-ee-orderid-map";//被评价人是雇员
    private static final String EVALUATE_ER_ORDERID_MAP = "evaluate-er-orderid-map";//被评价人是雇主
    private static final String ORDER_OPERATE_ORDERID_LIST = "orderoperate-orderid-list";//订单操作列表
    private static final String ORDER_OPERATE_OPERATEID_MAP = "operate-operateid-map";//订单操作详情

    //任务信息
    private static final String JOB_JOBID_MAP = "job-jobid-map";

    //服务信息
    private static final String SERVICE_SERVICEID_MAP = "service-serviceid-map";
    //用户创建的服务列表
    private static final String SERVICE_USERID_LIST = "service-userid-list";

    //用户信息
//    private static final String USER_USERID_EXPAND_MAP = "user-userid-expand-map";
    private static final String USER_USERID_PERSONAL_EXPAND_MAP = "user-userid-personal-expand-map";
    private static final String USER_USERID_ENTERPRESE_EXPAND_MAP = "user-userid-enterprise-expand-map";
    private static final String USER_USERID_MAP = "user-userid-map";

    //代理详情
    private static final String JOBAGENT_JOBID_MAP = "jobagent-jobid-map";

    //代理商的任務id和任務id的对应关系
    private static final String JOBAGENT_PARENTJOBID_STR = "jobagent-parentjobid-str";

    //附件详情
    private static final String ATTACHMENT_ID_MAP = "attachment-id-map";
    //附件详情
    private static final String ATTACHMENT_SERVICEID_LIST = "attachment-serviceid-list";

    //ES中存储的service的index
    private static final String ES_DLG_INDEX = PropertyUtil.getValue("es.service.index");
    //ES中存储的service的doc
    private static final String ES_SERVICE_TYPE = PropertyUtil.getValue("es.service.doc");
    //ES中存储的user的doc
    private static final String ES_USER_TYPE = PropertyUtil.getValue("es.user.doc");

    //邀请奖励状态
    private static final String INVITER_PHONE_ENTID_REWARD = "inviter-phone-entid-reward";
    //分享奖励状态
    private static final String SHARE_USERID_ENTID_REWARD = "share-userid-entid-reward";
    //邀请活动
    private static final String INVITATION_ACTIVITY = "invitation-activity";
    //邀请人手机号对应的企业列表
    private static final String INVITER_PHONE_ENTLIST = "inviter-phone-entlist";
    //双十一活动开始时间
    private static final String REWARD_ACTIVITY_START = Redis.get("personal.reward.activity.start");
    //双十一活动结束时间
    private static final String REWARD_ACTIVITY_END = Redis.get("personal.reward.activity.end");
    //企业的邀请关系
    private static final String RELATIONSHIP_INVITEE_ENTID = "relationship-invitee-entid";
    //企业的邀请关系
    private static final String SHARE_USERID_REWARD_COUNT = "share-userid-reward-count";

    //好评量
    private static final String USER_USERID_PRAISE_COUNT = "user-userid-praise-count";
    //差评量
    private static final String USER_USERID_NEGATIVE_COUNT = "user-userid-negative-count";
    //成交量
    private static final String USER_USERID_TRADING_COUNT = "user-userid-trading-count";
    //分享人的奖励上限
    private static final Integer SHARE_MAX_REWARD = Integer.valueOf(PropertyUtil.getValue("share.max.reward"));

    public static void main(String args[]) {
        // 创建链接
        LOGGER.info(PropertyUtil.getValue("canal.zkServers"));
        LOGGER.info(PropertyUtil.getValue("canal.client.destination"));
        LOGGER.info(PropertyUtil.getValue("canal.zkServers"));
        CanalConnector connector = CanalConnectors.newClusterConnector(PropertyUtil.getValue("canal.zkServers"),PropertyUtil.getValue("canal.client.destination"),PropertyUtil.getValue("canal.username"),PropertyUtil.getValue("canal.password"));
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
//            connector.subscribe("dlg\\,dlg.oj_user,dlg.oj_order,dlg.oj_order_time,dlg.oj_order_evaluate,dlg.oj_order_operate");
            connector.subscribe(PropertyUtil.getValue("canal.tables"));
            connector.rollback();
            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
//                    LOGGER.info("empty count : " + emptyCount);
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                    }
                } else {
                    emptyCount = 0;
                    manageRedisEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        } catch (Exception e){
            e.printStackTrace();
        }finally {
            connector.disconnect();
        }
    }

    private static void manageRedisEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }

            EventType eventType = rowChage.getEventType();
            LOGGER.info(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));

            for (RowData rowData : rowChage.getRowDatasList()) {
                try {
                    if (eventType == EventType.DELETE) {
                        String tableName = entry.getHeader().getTableName();
                        delete(rowData.getBeforeColumnsList(),tableName);
                        printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == EventType.INSERT) {
                        String tableName = entry.getHeader().getTableName();
                        insert(rowData, tableName);
                        printColumn(rowData.getAfterColumnsList());
                    } else if (eventType == EventType.UPDATE){
                        LOGGER.info("-------> before");
                        printColumn(rowData.getBeforeColumnsList());
                        LOGGER.info("-------> after");
                        printColumn(rowData.getAfterColumnsList());
                        String tableName = entry.getHeader().getTableName();
                        update(rowData.getBeforeColumnsList(),rowData.getAfterColumnsList(), tableName);
                    }
                }catch (Exception e){
                    errorLogger.error("ERROR",e);
                    errorLogger.error("-------> before");
                    printColumn(rowData.getBeforeColumnsList());
                    errorLogger.error("-------> after");
                    printColumn(rowData.getAfterColumnsList());
                    continue;
                }

            }
        }
    }

    private static void update(List<Column> beforeColumns, List<Column> afterColumns, String tableName) {
        if ("oj_order".equals(tableName)){//我接的零工
            updateOrder(beforeColumns, afterColumns);
        }else if("oj_order_time".equals(tableName)){//订单时间
            updateOrderTime(afterColumns);
        }else if("oj_job_task".equals(tableName)){//我发的零工
            updateJob(beforeColumns, afterColumns);
        }else if("oj_job_task_time".equals(tableName)){//任务时间
            updateJobTime(afterColumns);
        }else if("oj_user".equals(tableName)){//用户
            insertOrUpdateUser(afterColumns);
        }else if("oj_user_personal_attribute".equals(tableName)){//个人用户
            updateUserPersonal(afterColumns);
        }else if("oj_user_enterprise_attribute".equals(tableName)){//企业用户
            insertOrUpdateUserEnterprise(afterColumns);
        }else if("oj_order_evaluate".equals(tableName)){//订单评价
            updateOrderEvaluate(afterColumns);
        }else if("oj_order_operate".equals(tableName)){//订单操作记录
            updateOrderOperate(afterColumns);
        }else if("oj_job_service".equals(tableName)){
            updateJobService(afterColumns);
        }else if("oj_order_cancel".equals(tableName)){
            insertOrUpdateOrderCancel(afterColumns);
        }
    }

    private static void updateJobService(List<Column> columns) {
        String id="";
        String create_time="";
        String user_id="";
        String service_name="";
        String service_meter_unit="";
        Integer price=0;
        String advantage="";
        String start_time="";
        String end_time="";
        String work_day="";
        String x_coordinate="";
        String y_coordinate="";
        String tag="";
        String credit_count="36.5";
        String active = "";
        for (Column column:columns) {
            if ("id".equals(column.getName())) {
                id = column.getValue();
            }else if("active".equals(column.getName())){
                active = column.getValue();
            }else if("create_time".equals(column.getName())){
                create_time = column.getValue();
            }else if("user_id".equals(column.getName())){
                user_id = column.getValue();
            } else if("service_meter_unit".equals(column.getName())){
                service_meter_unit = column.getValue();
            }else if("price".equals(column.getName())){
                price = Integer.valueOf(StringUtils.isEmpty(column.getValue()) ? "0" : column.getValue());
            }else if("advantage".equals(column.getName())){
                advantage = column.getValue();
            }else if("start_time".equals(column.getName())){
                start_time = column.getValue();
            }else if("end_time".equals(column.getName())){
                end_time = column.getValue();
            }else if("work_day".equals(column.getName())){
                work_day = column.getValue();
            }else if("x_coordinate".equals(column.getName())){
                x_coordinate = column.getValue();
            }else if("y_coordinate".equals(column.getName())){
                y_coordinate = column.getValue();
            }else if("tag".equals(column.getName())){
                tag = column.getValue();
            }else if("service_name".equals(column.getName())){
                service_name = column.getValue();
            }
        }
        if ("0".equals(active)){//删除
            deleteJobService(columns);
        }else{
            String userType = Redis.hget(USER_USERID_MAP.replace("userid",user_id), "type");
            //服务详情放到redis
            for (Column column:columns) {
                Redis.hset(SERVICE_SERVICEID_MAP.replace("serviceid",id),column.getName(),column.getValue());
            }
            //查询用户的诚信值
            if ("PERSONAL".equals(userType)){
                credit_count = Redis.hget(USER_USERID_PERSONAL_EXPAND_MAP.replace("userid", user_id),"credit_count");
            }else{
                credit_count = Redis.hget(USER_USERID_ENTERPRESE_EXPAND_MAP.replace("userid", user_id),"credit_count");
            }

//            DBHelper helper = new DBHelper("select credit_count from oj_user_personal_attribute where user_id="+user_id+" limit 1");
//            try {
//                ResultSet ret = helper.pst.executeQuery();
//                while (ret.next()) {
//                    credit_count = ret.getString(1);
//                }
//            } catch (SQLException e) {
//                errorLogger.error("查询用户诚信值失败",e);
//            }
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            Map<String,Object> source = new HashMap<>();
            source.put("serviceid",id);
            source.put("userid",user_id);
            source.put("price",price);
            try {
                source.put("createtime",simpleDateFormat.parse(create_time));
            } catch (ParseException e) {
                source.put("createtime",new Date());
            }
            source.put("creditcount",Double.valueOf(credit_count));
            source.put("servicename",service_name);
            source.put("advantage",advantage);
            source.put("servicemeterunit",service_meter_unit);
            source.put("starttime",start_time);
            source.put("endtime",end_time);
            source.put("workday",work_day);
            source.put("tag",tag);
            Map<String,Double> xy = new HashMap<>();
            xy.put("lon",StringUtils.isEmpty(x_coordinate) ? Double.valueOf(0) : Double.valueOf(x_coordinate));
            xy.put("lat",StringUtils.isEmpty(y_coordinate) ? Double.valueOf(0) : Double.valueOf(y_coordinate));
            source.put("location",xy);

            try {
                ElasticsearchUtil.scrollSearchUpdate(ES_DLG_INDEX, ES_SERVICE_TYPE, "serviceid", id, source);
            } catch (Exception e) {
                ES_ERROR.error("修改服务到ES失败：",e);
            }
        }
    }

    private static void updateOrderOperate(List<Column> columns) {
        String id  = "";
        String active = "";
        for (Column column:columns){
            if ("id".equals(column.getName())){
                id = column.getValue();
            }if ("active".equals(column.getName())){
                active = column.getValue();
            }
        }
        if ("0".equals(active)){
            deleteOrderOperate(columns);
        }
        for (Column column : columns) {
            Redis.hset(ORDER_OPERATE_OPERATEID_MAP.replace("operateid",id),column.getName(),column.getValue());
        }
    }

    private static void updateJob(List<Column> beforeColumns, List<Column> afterColumns) {
        String active = "";//修改后的active
        String status = "";//修改后的状态
        String userId = "";//用户id
        String jobId = "";//任务id
        for (Column column:afterColumns) {
            if ("active".equals(column.getName()) && column.getUpdated()) {
                active = column.getValue();
            }else if("id".equals(column.getName())){
                jobId = column.getValue();
            }else if("status".equals(column.getName()) && column.getUpdated()){
                status = column.getValue();
            }else if("user_id".equals(column.getName())){
                userId = column.getValue();
            }
        }
        if (active.equals("0")){//任务改为无效
            deleteJob(afterColumns);
        }else if (active.equals("1")){//任务改为有效
            insertJob(afterColumns);
        }else {
            String beforeStatus = "";//修改前的状态
            if (StringUtils.isNotEmpty(status)) {//修改了任务状态
                for (Column column:beforeColumns) {
                    if("status".equals(column.getName())){
                        beforeStatus = column.getValue();
                        break;
                    }
                }
                //如果修改前和修改后的订单状态不一样
                if (!beforeStatus.equals(status)){
                    //我发的零工,状态有 进行中、已完成、待付款
                    if (beforeStatus.equals("20") || beforeStatus.equals("30") || beforeStatus.equals("40")) {
                        Redis.lrem(JOB_USERID_STATUS099_LIST.replace("99", beforeStatus).replace("userid", userId), 0, jobId);
                    }
                    if (status.equals("20") || status.equals("30") || status.equals("40")) {
                        Redis.lpush(JOB_USERID_STATUS099_LIST.replace("99", status).replace("userid", userId), jobId);
                    }
                }
            }
            //最新更新的任务要放到最前面
            Redis.lrem(JOB_USERID_STATUS099_LIST.replace("userid", userId), 0, jobId);
            Redis.lpush(JOB_USERID_STATUS099_LIST.replace("userid", userId), jobId);
            if (status.equals("20") || status.equals("30") || status.equals("40")) {
                Redis.lrem(JOB_USERID_STATUS099_LIST.replace("99", status).replace("userid", userId), 0, jobId);
                Redis.lpush(JOB_USERID_STATUS099_LIST.replace("99", status).replace("userid", userId), jobId);
            }
            //更新任务信息
            for (Column column:afterColumns){
                Redis.hset(JOB_JOBID_MAP.replace("jobid",jobId),column.getName(), column.getValue());
            }
        }
    }

    private static void updateOrder(List<Column> beforeColumns, List<Column> afterColumns) {
        String active = "";//修改后的active
        String orderId = "";//订单id
        String status = "";//修改后的status
        String isInProgress = "";//修改后的是否是进行中
        String employerId = "";//雇主id
        String employeeId = "";//雇员id
        String jobId = "";//任务id
        String total_price = "";
        String orderStatus="";
        for (Column column:afterColumns) {
            if("status".equals(column.getName())){
                orderStatus = column.getValue();
            }

            if ("active".equals(column.getName()) && column.getUpdated()) {
                active = column.getValue();
            }else if("id".equals(column.getName())){
                orderId = column.getValue();
            }else if("status".equals(column.getName()) && column.getUpdated()){
                status = column.getValue();
            }else if("is_in_progress".equals(column.getName())){
                isInProgress = column.getValue();
            }else if("employee_id".equals(column.getName())){
                employeeId = column.getValue();
            }else if("job_id".equals(column.getName())){
                jobId = column.getValue();
            }else if("employer_id".equals(column.getName())){
                employerId = column.getValue();
            }else if("total_price".equals(column.getName())){
                total_price = column.getValue();
            }
        }
        String jobOrderVal = orderId+"-"+employeeId;
        String jobOrderAfterKeyStatus = getJobOrderStatus(status);

        String parent_job_id = Redis.hget(JOB_JOBID_MAP.replace("jobid",jobId),"parent_job_id");

        if (active.equals("0")){
            deleteOrder(afterColumns);
        }else if (active.equals("1")){
            insertOrder(afterColumns);
        }else {//没更新active字段
            LOGGER.info("active:"+active);
            String redisOrderAfterKeyStatus = "";//修改后的redis中key的status
            String redisOrderBeforeKeyStatus = "";//修改前的redis中key的status
            String beforeStatus = "";//修改前的状态
            if (StringUtils.isNotEmpty(status)) {//修改了订单状态

                String beforeIsInProgress = "";//修改前的是否是进行中
                for (Column column:beforeColumns) {
                    if("status".equals(column.getName())){
                        beforeStatus = column.getValue();
                    }else if("is_in_progress".equals(column.getName())){
                        beforeIsInProgress = column.getValue();
                    }
                }


                if (beforeIsInProgress.equals("1") && !beforeStatus.equals("30") && !beforeStatus.equals("40")){
                    redisOrderBeforeKeyStatus = "020";
                }else if (beforeStatus.equals("30")){
                    redisOrderBeforeKeyStatus = "030";
                }else if (beforeStatus.equals("40")){
                    redisOrderBeforeKeyStatus = "040";
                }else if (status.equals("50")){
                    redisOrderBeforeKeyStatus = "050";
                }

                if (isInProgress.equals("1") && !status.equals("30") && !status.equals("40")){
                    redisOrderAfterKeyStatus = "020";
                }else if (status.equals("30")){
                    redisOrderAfterKeyStatus = "030";
                }else if (status.equals("40")){
                    redisOrderAfterKeyStatus = "040";
                }else if (status.equals("50")){
                    redisOrderAfterKeyStatus = "050";
                }

                //joborder 的状态
                String jobOrderBeforeKeyStatus = getJobOrderStatus(beforeStatus);


                //如果修改前和修改后的订单状态不一样
                if (!redisOrderBeforeKeyStatus.equals(redisOrderAfterKeyStatus)){
                    //我接的零工没有"已取消"的分类
                    if (StringUtils.isNotEmpty(redisOrderBeforeKeyStatus) && !redisOrderBeforeKeyStatus.equals("50")) {
                        Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderBeforeKeyStatus).replace("userid", employeeId), 0, orderId);
                    }
                    //我接的零工没有"已取消"的分类
                    if (StringUtils.isNotEmpty(redisOrderAfterKeyStatus) && !redisOrderAfterKeyStatus.equals("50")) {
                        Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("userid", employeeId), orderId);
                    }
                }
                LOGGER.info("jobOrderBeforeKeyStatus:"+jobOrderBeforeKeyStatus);
                LOGGER.info("jobOrderAfterKeyStatus:"+jobOrderAfterKeyStatus);
                if (!jobOrderBeforeKeyStatus.equals(jobOrderAfterKeyStatus)){
                    if (StringUtils.isNotEmpty(parent_job_id)){
                        Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderBeforeKeyStatus).replace("jobid", parent_job_id), 0,jobOrderVal);
                        Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderAfterKeyStatus).replace("jobid", parent_job_id), jobOrderVal);
                    }
                    Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderBeforeKeyStatus).replace("jobid", jobId), 0, jobOrderVal);
                    Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderAfterKeyStatus).replace("jobid", jobId), jobOrderVal);

                }

                //成交量
                if ("40".equals(orderStatus)){
                    if (!"40".equals(beforeStatus)){
                        Redis.incr(USER_USERID_TRADING_COUNT.replace("userid", employerId));
                    }
                }else{
                    if ("40".equals(beforeStatus)){
                        Redis.decr(USER_USERID_TRADING_COUNT.replace("userid", employerId));
                    }
                }
            }

            //将最新更新的放在最前面
            Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("userid", employeeId), 0, orderId);
            Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("userid", employeeId), orderId);
            Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("jobid", jobId), 0, jobOrderVal);
            Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("jobid", jobId), jobOrderVal);
            if (StringUtils.isNotEmpty(parent_job_id)){
                Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("jobid", parent_job_id), 0, jobOrderVal);
                Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("jobid", parent_job_id), jobOrderVal);
            }

            //order的状态如果不为空，则将最新更新的放在最前面
            if (StringUtils.isNotEmpty(redisOrderAfterKeyStatus)){
                Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("userid", employeeId), 0, orderId);
                Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("userid", employeeId), orderId);
            }
            //joborder的状态如果不为空，则将最新更新的放在最前面
            if (StringUtils.isNotEmpty(jobOrderAfterKeyStatus)){
                Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("jobid", jobId), 0, jobOrderVal);
                Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("jobid", jobId), jobOrderVal);
                if (StringUtils.isNotEmpty(parent_job_id)) {
                    Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("jobid", parent_job_id), 0, jobOrderVal);
                    Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", redisOrderAfterKeyStatus).replace("jobid", parent_job_id), jobOrderVal);
                }
            }

            //更新订单详情
            for (Column column : afterColumns) {
                Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), column.getName(), column.getValue());
            }



            try {
                inviterReward(jobId, orderId, orderStatus, employerId, total_price);
            }catch (Exception e){
                e.printStackTrace();
                errorLogger.error("双十一活动奖励失败",e);
            }
        }
    }

    private static void inviterReward(String jobId, String orderId,String orderStatus, String employerId, String total_price){

        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date rewardActStart = new Date();
        Date rewardActEnd = new Date();
        try {
            rewardActStart = simpleDateFormat.parse(REWARD_ACTIVITY_START);
            rewardActEnd = simpleDateFormat.parse(REWARD_ACTIVITY_END);
        }catch (Exception e){
            errorLogger.error("活动时间格式转换失败",e);
        }

        if ("40".equals(orderStatus) && StringUtils.isNotEmpty(total_price) && Integer.valueOf(total_price)>=500){//已完成并且金额>=50
            LOGGER.info("-----1------");
            String userType = Redis.hget(USER_USERID_MAP.replace("userid",employerId),"type");
            if ("ENTERPRISE".equals(userType)){//企业
                LOGGER.info("-----2------");
                Map<String,String> ship = Redis.hgetAll(RELATIONSHIP_INVITEE_ENTID.replace("entid",employerId));
                if (ship!=null && StringUtils.isNotEmpty(ship.get("id")) && !"1".equals(ship.get("activityIsEnd"))){
                    LOGGER.info("-----3------");
                    //邀请人手机号
                    String inviterPhone = ship.get("inviterPhone");
                    //还没奖励过邀请人
                    if (!"1".equals(Redis.get(INVITER_PHONE_ENTID_REWARD.replace("phone",inviterPhone).replace("entid",employerId)))){
                        //分享人user_phone
                        String srcUserPhone = ship.get("srcUserPhone");
                        String shareRewardCount = Redis.get(SHARE_USERID_REWARD_COUNT.replace("userid", srcUserPhone));
                        Integer src = StringUtils.isEmpty(shareRewardCount) ? 0 : Integer.valueOf(shareRewardCount);

                        LOGGER.info("-----4------" + Redis.get(INVITER_PHONE_ENTID_REWARD.replace("phone", inviterPhone)));
                        //企业名称
                        String name = Redis.hget(USER_USERID_ENTERPRESE_EXPAND_MAP.replace("userid", employerId), "name");

                        //邀请人类型
                        String inviterType = ship.get("inviterType");
                        //邀请人奖励
                        String inviterAward = ship.get("inviterAward");
                        //分享人奖励
                        String srcUserAward = ship.get("srcUserAward");

                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put("order_id", orderId);
                        jsonObject.put("order_total_price", total_price);
                        jsonObject.put("employer_id", employerId);
                        jsonObject.put("employer_type", userType);
                        jsonObject.put("employer_name", name);
                        jsonObject.put("inviter_phone", inviterPhone);
                        jsonObject.put("inviter_type", inviterType);
                        jsonObject.put("inviter_award", inviterAward);
                        if (src<SHARE_MAX_REWARD) {//分享人的奖励有上限
                            jsonObject.put("src_user_phone", srcUserPhone);
                            jsonObject.put("src_user_award", srcUserAward);
                            Redis.set(SHARE_USERID_REWARD_COUNT.replace("userid", srcUserPhone), String.valueOf(src + 1));
                        }
                        Redis.lpush(INVITATION_ACTIVITY, JSON.toJSONString(jsonObject));
                        Redis.set(INVITER_PHONE_ENTID_REWARD.replace("phone", inviterPhone).replace("entid", employerId), "1");
                    }
                }
            }else {//个人
                LOGGER.info("=========个人==========");
                String inviterPhone = Redis.hget(USER_USERID_MAP.replace("userid",employerId), "phone");
                //雇主名称
                String personalName = Redis.hget(USER_USERID_ENTERPRESE_EXPAND_MAP.replace("userid",employerId),"name");
                SimpleDateFormat simpleDateFormat1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String jobCreateTime = Redis.hget(JOB_JOBID_MAP.replace("jobid",jobId), "create_time");
                Date jobTime = new Date();
                try {
                    jobTime = StringUtils.isEmpty(jobCreateTime) ? new Date() : simpleDateFormat1.parse(jobCreateTime);
                } catch (ParseException e) {
                    errorLogger.error("任务创建时间日期格式转换失败",e);
                }
                //发单时间在活动期间
                if (rewardActStart.before(jobTime) && rewardActEnd.after(jobTime)){
                    LOGGER.info("=========1==========");
                    List<String> inviterEntList = Redis.lAll(INVITER_PHONE_ENTLIST.replace("phone",inviterPhone));
                    for (String ent:inviterEntList){
                        LOGGER.info("=========2 for==========");
                        //企业认证状态
                        String audit_status = Redis.hget(USER_USERID_MAP.replace("userid",ent),"audit_status");
                        if ("2".equals(audit_status)){//企业认证通过了
                            LOGGER.info("=========3 audit_status=========="+audit_status);
                            Map<String,String> ship = Redis.hgetAll(RELATIONSHIP_INVITEE_ENTID.replace("entid",ent));
                            String activityIsEnd = ship.get("activityIsEnd");
                            if ("1".equals(activityIsEnd)){
                                continue;
                            }


                            if (!"1".equals(Redis.get(INVITER_PHONE_ENTID_REWARD.replace("phone",inviterPhone).replace("entid",ent)))){
                                LOGGER.info("=========6 还没奖励给邀请人==========");

                                //邀请人类型
                                String inviterType = ship.get("inviterType");
                                //邀请人奖励
                                String inviterAward = ship.get("inviterAward");
                                JSONObject jsonObject = new JSONObject();
                                jsonObject.put("order_id",orderId);
                                jsonObject.put("order_total_price",total_price);
                                jsonObject.put("employer_id",employerId);
                                jsonObject.put("employer_type",userType);
                                jsonObject.put("employer_name",personalName);
                                jsonObject.put("inviter_phone",inviterPhone);
                                jsonObject.put("inviter_type",inviterType);
                                jsonObject.put("inviter_award",inviterAward);

                                Redis.lpush(INVITATION_ACTIVITY, JSON.toJSONString(jsonObject));
                                Redis.set(INVITER_PHONE_ENTID_REWARD.replace("phone",inviterPhone).replace("entid",ent),"1");
                            }


                            String shareReward = Redis.get(SHARE_USERID_ENTID_REWARD.replace("userid",ship.get("srcUserPhone")).replace("entid",ent));
                            //奖励分享人
                            if (!"1".equals(shareReward)){
                                LOGGER.info("=========4 shareReward=========="+shareReward);
                                //分享人user_phone
                                String srcUserPhone = ship.get("srcUserPhone");
                                String shareRewardCount = Redis.get(SHARE_USERID_REWARD_COUNT.replace("userid",srcUserPhone));
                                Integer src = StringUtils.isEmpty(shareRewardCount) ? 0 : Integer.valueOf(shareRewardCount);
                                if (src<SHARE_MAX_REWARD){//分享人的奖励有上限
                                    LOGGER.info("=========5 shareReward src=========="+src);
                                    //雇主名称
                                    String erName = Redis.hget(USER_USERID_PERSONAL_EXPAND_MAP.replace("userid",employerId),"name");
                                    //企业名称
                                    String name = Redis.hget(USER_USERID_ENTERPRESE_EXPAND_MAP.replace("userid",ent),"name");

                                    //分享人奖励
                                    String srcUserAward = ship.get("srcUserAward");

                                    JSONObject jsonObject = new JSONObject();
                                    jsonObject.put("order_id",orderId);
                                    jsonObject.put("order_total_price",total_price);
                                    jsonObject.put("employer_id",employerId);
                                    jsonObject.put("employer_type",userType);
                                    jsonObject.put("employer_name",erName);
                                    jsonObject.put("ent_id",ent);
                                    jsonObject.put("ent_name",name);
                                    jsonObject.put("src_user_phone",srcUserPhone);
                                    jsonObject.put("src_user_award",srcUserAward);

                                    Redis.lpush(INVITATION_ACTIVITY, JSON.toJSONString(jsonObject));
                                    Redis.set(SHARE_USERID_ENTID_REWARD.replace("userid", srcUserPhone).replace("entid",ent),"1");
                                    Redis.set(SHARE_USERID_REWARD_COUNT.replace("userid",srcUserPhone), String.valueOf(src+1));
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    private static String getJobOrderStatus(String status){
//        30,22,10,21(20),50(51),40,6(8),7(9)
//        订单状态(6.有人雇佣 7.拒绝雇佣 8等待雇员同意 9雇员拒绝,10.等待同意,11.拒绝接单,
//        20.等待开工,21.正在干活中,22.等待验收,30.待收款,40.已完成,50.已取消,60.强行终止)',
        String result = "";
        if (StringUtils.isNotEmpty(status)){
            if (Integer.valueOf(status)<10){
                result = "00"+status;
            }else{
                result = "0"+status;
            }
        }

//        if ("6".equals(status)){
//            result = "006";
//        }else if("7".equals(status)){
//            result = "007";
//        }else if("8".equals(status)){
//            result = "008";
//        }else if("9".equals(status)){
//            result = "009";
//        }else if("10".equals(status)){
//            result = "010";
//        }else if("11".equals(status)){
//            result = "011";
//        }else if("20".equals(status)){
//            result = "020";
//        }else if("21".equals(status)){
//            result = "021";
//        }else if("22".equals(status)){
//            result = "022";
//        }else if("30".equals(status)){
//            result = "030";
//        }else if("40".equals(status)){
//            result = "040";
//        }else if("50".equals(status)){
//            result = "050";
//        }else if("51".equals(status)){
//            result = "051";
//        }else if("60".equals(status)){
//            result = "060";
//        }
        return result;
    }

    private static void delete(List<Column> columns, String tableName) {
        if ("oj_order".equals(tableName)){//删除订单
            deleteOrder(columns);
        }else if("oj_job_task".equals(tableName)){//删除任务
            deleteJob(columns);
        }else if("oj_user".equals(tableName)){//删除用户
            deleteUser(columns);
        }else if("oj_order_evaluate".equals(tableName)){//订单评价
            deleteOrderEvaluate(columns);
        }else if("oj_order_operate".equals(tableName)){//订单操作记录
            deleteOrderOperate(columns);
        }else if("oj_job_task_agent".equals(tableName)){//任务代理
            deleteJobAgentMap(columns);
        }else if("oj_order_cancel".equals(tableName)){//取消订单
            deleteOrderCancel(columns);
        }else if("oj_job_service".equals(tableName)){//服务
            deleteJobService(columns);
        }else if("oj_attachment".equals(tableName)){//附件
            deleteAttachement(columns);
        }else if("oj_job_service_attachment".equals(tableName)){//服务于附件的关系
            deleteServiceAttachment(columns);
        }
    }

    private static void deleteAttachement(List<Column> columns) {
        String id = "";
        for (Column column:columns) {
            if ("id".equals(column.getName())) {
                id = column.getValue();
                break;
            }
        }
        Redis.del(ATTACHMENT_ID_MAP.replace("id",id));
    }

    private static void deleteJobService(List<Column> columns) {
        String serviceId = "";
        String userId = "";
        for (Column column:columns) {
            if ("id".equals(column.getName())) {
                serviceId = column.getValue();
            }else if ("user_id".equals(column.getName())) {
                userId = column.getValue();
            }
        }
        //删除服务详情
        Redis.del(SERVICE_SERVICEID_MAP.replace("serviceid",serviceId));
        //删除"用户创建的服务"列表
        Redis.lrem(SERVICE_USERID_LIST.replace("userid",userId),0,serviceId);
        try {
            ElasticsearchUtil.scrollSearchDelete(ES_DLG_INDEX, ES_SERVICE_TYPE, "serviceid",serviceId);
        } catch (Exception e) {
            ES_ERROR.error("从ES中删除服务失败：",e);
        }
    }

    private static void deleteOrderCancel(List<Column> columns) {
        String orderId = "";
        for (Column column:columns) {
            if("order_id".equals(column.getName())){
                orderId = column.getValue();
                break;
            }
        }
        Redis.del(ORDER_ORDERID_CANCEL_MAP.replace("orderid",orderId));
    }

    private static void deleteOrderOperate(List<Column> columns) {
        String orderId  = "";
        String id  = "";
        for (Column column:columns){
            if ("id".equals(column.getName())){
                id = column.getValue();
            }else if ("order_id".equals(column.getName())){
                orderId = column.getValue();
            }
        }
        if (StringUtils.isNotEmpty(orderId)) {
            Redis.lrem(ORDER_OPERATE_ORDERID_LIST.replace("orderid", orderId),0,id);
            Redis.del(ORDER_OPERATE_OPERATEID_MAP.replace("operateid",id));
        }
    }

    private static void deleteOrderEvaluate(List<Column> columns) {
        String orderId = "";
        String accepter_user_id = "";
        String evaluate_level = "";
        for (Column column:columns){
            if ("order_id".equals(column.getName())){
                orderId = column.getValue();
            }else if("accepter_user_id".equals(column.getName())){//被评价人用户id
                accepter_user_id = column.getValue();
            }else if("evaluate_level".equals(column.getName())){
                evaluate_level = column.getValue();
            }
        }
        String employeeId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid", orderId), "employee_id");
        String employerId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid", orderId), "employer_id");

        if (StringUtils.isNotEmpty(evaluate_level) && Integer.valueOf(evaluate_level)>=5){
            Redis.incr(USER_USERID_PRAISE_COUNT.replace("userid", accepter_user_id));
        }else{
            Redis.incr(USER_USERID_NEGATIVE_COUNT.replace("userid", accepter_user_id));
        }

        if (StringUtils.isNotEmpty(orderId)) {
            if (employeeId.equals(accepter_user_id)) {//被评价人是雇员
                Redis.del(EVALUATE_EE_ORDERID_MAP.replace("orderid", orderId));
            } else if (employerId.equals(accepter_user_id)) {//被评价人是雇主
                Redis.del(EVALUATE_ER_ORDERID_MAP.replace("orderid", orderId));
            }
        }
    }

    private static void deleteUser(List<Column> columns) {
        String userId = "";
        for (Column column:columns){
            if (StringUtils.isNotEmpty(userId))
                break;
            if ("id".equals(column.getName())){
                userId = column.getValue();
            }
        }
        Redis.del(USER_USERID_PERSONAL_EXPAND_MAP.replace("userid",userId));//用户扩展信息
        Redis.del(USER_USERID_ENTERPRESE_EXPAND_MAP.replace("userid",userId));//用户扩展信息
        Redis.del(USER_USERID_MAP.replace("userid",userId));//用户信息
    }

    private static void deleteJob(List<Column> columns) {
        String status = "";
        String userId = "";
        String jobId = "";
        for (Column column : columns) {
            if ("status".equals(column.getName())){
                status = column.getValue();
            }else if ("user_id".equals(column.getName())){
                userId = column.getValue();
            }else if ("id".equals(column.getName())){
                jobId = column.getValue();
            }
        }
        String replaceStatus = "status0"+status;
        Redis.lrem(JOB_USERID_STATUS099_LIST.replace("userid",userId),0,jobId);//用户发布的任务
        //用户发布的任务
        Redis.lrem(JOB_USERID_STATUS099_LIST.replace("status099",replaceStatus).replace("userid",userId),0,jobId);

        String[] jobOrderStatusAll = JOBORDER_STATUS_ALL.split(",");
        for (String jobOrderStatus:jobOrderStatusAll){
            Redis.del(JOBORDER_JOBID_STATUS099_LIST.replace("099",jobOrderStatus).replace("jobid",jobId));
        }
        Redis.del(JOB_JOBID_MAP.replace("jobid",jobId));//任务信息
//        Redis.del(JOB_JOBID_EXPAND_MAP.replace("jobid",jobId));//任务扩展信息
    }

    private static void deleteOrder(List<Column> columns) {
        String orderId = "";
        String jobId = "";
        String employeeId = "";
        for (Column column : columns) {
            if ("id".equals(column.getName())){
                orderId = column.getValue();
            }else if("employee_id".equals(column.getName())){
                employeeId = column.getValue();
            }else if("job_id".equals(column.getName())){
                jobId = column.getValue();
            }
        }

        String parent_job_id = Redis.hget(JOB_JOBID_MAP.replace("jobid",jobId),"parent_job_id");
        if (StringUtils.isNotEmpty(parent_job_id)){
            Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("jobid", parent_job_id), 0,orderId + "-" + employeeId);
        }
        String jobOrderVal = orderId+"-"+employeeId;

        Redis.del(ORDER_ORDERID_MAP.replace("orderid",orderId));//订单信息
        Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("userid",employeeId),0, orderId);//用户接单列表
        Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("jobid",jobId),0,jobOrderVal);//任务对应的订单列表
        String[] jobOrderStatusAll = JOBORDER_STATUS_ALL.split(",");
        for (String jobOrderStatus:jobOrderStatusAll){
            Redis.del(JOBORDER_JOBID_STATUS099_LIST.replace("099",jobOrderStatus).replace("jobid",jobId));
        }
//        Redis.del(EVALUATE_EE_ORDERID_MAP.replace("orderid",orderId));//被评价人是雇员
//        Redis.del(EVALUATE_ER_ORDERID_MAP.replace("orderid",orderId));//被评价人是雇主
//        Redis.del(ORDER_OPERATE_ORDERID_LIST.replace("orderid",orderId));//被评价人是雇主
//        Redis.del(EVALUATE_ER_ORDERID_MAP.replace("orderid",orderId));//被评价人是雇主
    }

    private static void insert(RowData rowData, String tableName){
        List<Column> columns = rowData.getAfterColumnsList();
        if ("oj_order".equals(tableName)){//我接的零工
            insertOrder(columns);
        }else if("oj_order_time".equals(tableName)){//订单时间
            insertOrderTime(columns);
        }else if("oj_job_task".equals(tableName)){//我发的零工
            insertJob(columns);
        }else if("oj_job_task_time".equals(tableName)){//任务时间
            insertJobTime(columns);
        }else if("oj_user".equals(tableName)){//用户
            insertOrUpdateUser(columns);
        }else if("oj_user_personal_attribute".equals(tableName)){//个人用户
            insertUserPersonal(columns);
        }else if("oj_user_enterprise_attribute".equals(tableName)){//企业用户
            insertOrUpdateUserEnterprise(columns);
        }else if("oj_order_evaluate".equals(tableName)){//订单评价
            insertOrderEvaluate(rowData.getAfterColumnsList());
        }else if("oj_order_operate".equals(tableName)){//订单操作记录
            insertOrderOperate(rowData.getAfterColumnsList());
        }else if("oj_job_task_agent".equals(tableName)){
            insertOrUpdateJobTaskAgent(rowData.getAfterColumnsList());
        }else if("oj_order_cancel".equals(tableName)){
            insertOrUpdateOrderCancel(rowData.getAfterColumnsList());
        }else if("oj_job_service".equals(tableName)){
            insertJobService(rowData.getAfterColumnsList());
        }else if("oj_attachment".equals(tableName)){
            insertOrUpdateAttachment(rowData.getAfterColumnsList());
        }else if("oj_job_service_attachment".equals(tableName)){
            insertServiceAttachment(columns);
        }

    }

    private static void insertServiceAttachment(List<Column> columns) {
        String serviceId = "";
        String attachmentId = "";
        for (Column column:columns) {
            if ("attachment_id".equals(column.getName())){
                attachmentId = column.getValue();
            }else if ("job_service_id".equals(column.getName())) {
                serviceId=column.getValue();
            }
        }
        Redis.lpush(ATTACHMENT_SERVICEID_LIST.replace("serviceid",serviceId),attachmentId);
    }

    private static void deleteServiceAttachment(List<Column> columns) {
        String serviceId = "";
        String attachmentId = "";
        for (Column column:columns) {
            if ("attachment_id".equals(column.getName())){
                attachmentId = column.getValue();
            }else if ("job_service_id".equals(column.getName())) {
                serviceId=column.getValue();
            }
        }
        Redis.lrem(ATTACHMENT_SERVICEID_LIST.replace("serviceid",serviceId),0,attachmentId);
    }

    private static void insertOrUpdateAttachment(List<Column> columns) {
        String id ="";
        String active = "";
        for (Column column:columns) {
            if ("id".equals(column.getName())) {
                id=column.getValue();
            }else if("active".equals(column.getName())){
                active = column.getValue();
            }
        }
        if ("0".equals(active)){
            Redis.del(ATTACHMENT_ID_MAP.replace("id",id));
        }else{
            for (Column column:columns) {
                Redis.hset(ATTACHMENT_ID_MAP.replace("id",id),column.getName(),column.getValue());
            }
        }
    }

    private static void insertJobService(List<Column> columns) {
        String id="";
        String create_time="";
        String user_id="";
        String service_name="";
        String service_meter_unit="";
        Integer price=0;
        String advantage="";
        String start_time="";
        String end_time="";
        String work_day="";
        String x_coordinate="";
        String y_coordinate="";
        String tag="";
        String credit_count="36.5";
        for (Column column:columns) {
            if ("id".equals(column.getName())) {
                id = column.getValue();
            }else if("create_time".equals(column.getName())){
                create_time = column.getValue();
            }else if("user_id".equals(column.getName())){
                user_id = column.getValue();
            } else if("service_meter_unit".equals(column.getName())){
                service_meter_unit = column.getValue();
            }else if("price".equals(column.getName())){
                price = Integer.valueOf(StringUtils.isEmpty(column.getValue()) ? "0" : column.getValue());
            }else if("advantage".equals(column.getName())){
                advantage = column.getValue();
            }else if("start_time".equals(column.getName())){
                start_time = column.getValue();
            }else if("end_time".equals(column.getName())){
                end_time = column.getValue();
            }else if("work_day".equals(column.getName())){
                work_day = column.getValue();
            }else if("x_coordinate".equals(column.getName())){
                x_coordinate = column.getValue();
            }else if("y_coordinate".equals(column.getName())){
                y_coordinate = column.getValue();
            }else if("tag".equals(column.getName())){
                tag = column.getValue();
            }else if("service_name".equals(column.getName())){
                service_name = column.getValue();
            }
        }
        String userType = Redis.hget(USER_USERID_MAP.replace("userid",user_id), "type");
        if ("PERSONAL".equals(userType)){
            credit_count = Redis.hget(USER_USERID_PERSONAL_EXPAND_MAP.replace("userid", user_id),"credit_count");
        }else {
            credit_count = Redis.hget(USER_USERID_ENTERPRESE_EXPAND_MAP.replace("userid", user_id),"credit_count");
        }

//        DBHelper helper = new DBHelper("select credit_count from oj_user_personal_attribute where user_id="+user_id+" limit 1");
//        try {
//            ResultSet ret = helper.pst.executeQuery();
//            while (ret.next()) {
//                credit_count = ret.getString(1);
//                System.out.println(id);
//            }//显示数据
//        } catch (SQLException e) {
//            errorLogger.error("sql error",e);
//        }

        //"用户创建的服务"列表，放到redis
        Redis.lpush(SERVICE_USERID_LIST.replace("userid",user_id), id);
        //服务详情，放到redis
        for (Column column:columns) {
            Redis.hset(SERVICE_SERVICEID_MAP.replace("serviceid",id),column.getName(),column.getValue());
        }
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Map<String,Object> source = new HashMap<>();
        source.put("serviceid",id);
        source.put("userid",user_id);
        source.put("price",price);
        try {
            source.put("createtime",simpleDateFormat.parse(create_time));
        } catch (ParseException e) {
            source.put("createtime",new Date());
        }
        source.put("creditcount",Double.valueOf(credit_count));
        source.put("servicename",service_name);
        source.put("advantage",advantage);
        source.put("servicemeterunit",service_meter_unit);
        source.put("starttime",start_time);
        source.put("endtime",end_time);
        source.put("workday",work_day);
        source.put("tag",tag);
        Map<String,Double> xy = new HashMap<>();
        xy.put("lon",StringUtils.isEmpty(x_coordinate) ? Double.valueOf(0) : Double.valueOf(x_coordinate));
        xy.put("lat",StringUtils.isEmpty(y_coordinate) ? Double.valueOf(0) : Double.valueOf(y_coordinate));
        source.put("location",xy);

        try {
            TransportClient client = ElasticsearchUtil.getClient();
            client.prepareIndex(ES_DLG_INDEX, ES_SERVICE_TYPE).setId(id).setSource(source).get();
        } catch (Exception e) {
            ES_ERROR.error("保存服务到ES失败：",e);
        }
    }

    private static void insertOrUpdateOrderCancel(List<Column> columns) {
        String orderId = "";
        String active = "";
        for (Column column:columns) {
            if("order_id".equals(column.getName())){
                orderId = column.getValue();
            }else if("active".equals(column.getName())){
                active = column.getValue();
            }
        }
        if ("0".equals(active)){
            deleteOrderCancel(columns);
        }else{
            for (Column column:columns) {
                Redis.hset(ORDER_ORDERID_CANCEL_MAP.replace("orderid",orderId), column.getName(), column.getValue());
            }
//            Redis.hset(ORDER_ORDERID_MAP.replace("orderid",orderId),"cancel_cause", cancelCause);
        }
    }

    private static void insertOrUpdateJobTaskAgent(List<Column> columns) {
        String jobId  = "";
        String active  = "";
        String parent_job_id = "";
        for (Column column:columns){
            if ("job_id".equals(column.getName())){
                jobId = column.getValue();
            }else if("active".equals(column.getName())){
                active = column.getValue();
            }else if("parent_job_id".equals(column.getName())){
                parent_job_id = column.getValue();
            }
        }
        if ("0".equals(active)){
            deleteJobAgentMap(columns);
        }else {
            if (StringUtils.isNotEmpty(jobId)) {
                for (Column column : columns) {
                    Redis.hset(JOBAGENT_JOBID_MAP.replace("jobid", jobId), column.getName(), column.getValue());
                }
                if (StringUtils.isNotEmpty(parent_job_id)) {
                    Redis.set(JOBAGENT_PARENTJOBID_STR.replace("parentjobid", parent_job_id), jobId);
                }
            }
        }
    }

    private static void deleteJobAgentMap(List<Column> columns) {
        String jobId  = "";
        String parent_job_id = "";
        for (Column column:columns){
            if ("job_id".equals(column.getName())){
                jobId = column.getValue();
            }else if("parent_job_id".equals(column.getName())){
                parent_job_id = column.getValue();
            }
        }
        Redis.del(JOBAGENT_PARENTJOBID_STR.replace("parentjobid", parent_job_id));
        Redis.del(JOBAGENT_JOBID_MAP.replace("jobid", jobId));
    }

    private static void insertOrderOperate(List<Column> columns) {
        String orderId  = "";
        String id  = "";
        for (Column column:columns){
            if ("id".equals(column.getName())){
                id = column.getValue();
            }else if ("order_id".equals(column.getName())){
                orderId = column.getValue();
            }
        }
        if (StringUtils.isNotEmpty(orderId)) {
            Redis.lpush(ORDER_OPERATE_ORDERID_LIST.replace("orderid", orderId),id);
            for (Column column : columns) {
                Redis.hset(ORDER_OPERATE_OPERATEID_MAP.replace("operateid",id),column.getName(),column.getValue());
            }
        }
    }

    private static void insertOrderEvaluate(List<Column> columns) {
        String orderId = "";
        String active = "";
        String accepter_user_id = "";
        String evaluate_level = "";
        for (Column column:columns){
            if ("order_id".equals(column.getName())){
                orderId = column.getValue();
            }else if("active".equals(column.getName())){//修改了active字段
                active = column.getValue();
            }else if("accepter_user_id".equals(column.getName())){//被评价人用户id
                accepter_user_id = column.getValue();
            } else if ("evaluate_level".equals(column.getName())) {
                evaluate_level = column.getValue();
            }
        }

        //修改用户的好评数量 or 差评数量
        if (StringUtils.isNotEmpty(evaluate_level) && Integer.valueOf(evaluate_level)>=5){
//            String praise_count = Redis.hget(USER_USERID_EXPAND_MAP.replace("userid", accepter_user_id),"praise_count");
            Redis.incr(USER_USERID_PRAISE_COUNT.replace("userid", accepter_user_id));
        }else{
//            String negative_count = Redis.hget(USER_USERID_EXPAND_MAP.replace("userid", accepter_user_id),"negative_count");
            Redis.incr(USER_USERID_NEGATIVE_COUNT.replace("userid", accepter_user_id));
        }

        //查询雇员id
        String employeeId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid", orderId), "employee_id");
        //查询雇主id
        String employerId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid", orderId), "employer_id");

        if (StringUtils.isNotEmpty(orderId)) {
            if ("0".equals(active)) {//删除评价
                if (employeeId.equals(accepter_user_id)) {//被评价人是雇员
                    Redis.del(EVALUATE_EE_ORDERID_MAP.replace("orderid", orderId));
                }else if (employerId.equals(accepter_user_id)) {//被评价人是雇主
                    Redis.del(EVALUATE_ER_ORDERID_MAP.replace("orderid", orderId));
                }
            } else if ("1".equals(active)) {//将评价改为有效
                String key = "";
                if (employeeId.equals(accepter_user_id)){//被评价人是雇员
                    key = EVALUATE_EE_ORDERID_MAP;
                }else if (employerId.equals(accepter_user_id)){//被评价人是雇主
                    key = EVALUATE_ER_ORDERID_MAP;
                }
                if (StringUtils.isNotEmpty(key)){
                    Redis.hset(key.replace("orderid", orderId), "evaluate_level", evaluate_level);
                }
            }
        }
    }

    private static void updateOrderEvaluate(List<Column> columns) {
        String orderId = "";
        String active = "";
        String accepter_user_id = "";
        String evaluate_level = "";
        for (Column column:columns){
            if ("order_id".equals(column.getName())){
                orderId = column.getValue();
            }else if("active".equals(column.getName())){//修改了active字段
                active = column.getValue();
            }else if("accepter_user_id".equals(column.getName())){//被评价人用户id
                accepter_user_id = column.getValue();
            } else if ("evaluate_level".equals(column.getName())) {
                evaluate_level = column.getValue();
            }
        }
        if ("0".equals(active)){
            deleteOrderEvaluate(columns);
        }else {
            //查询雇员id
            String employeeId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid", orderId), "employee_id");
            //查询雇主id
            String employerId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid", orderId), "employer_id");

            if (StringUtils.isNotEmpty(orderId)) {
                if ("0".equals(active)) {//删除评价
                    if (employeeId.equals(accepter_user_id)) {//被评价人是雇员
                        Redis.del(EVALUATE_EE_ORDERID_MAP.replace("orderid", orderId));
                    }else if (employerId.equals(accepter_user_id)) {//被评价人是雇主
                        Redis.del(EVALUATE_ER_ORDERID_MAP.replace("orderid", orderId));
                    }
                } else if ("1".equals(active)) {//将评价改为有效
                    String key = "";
                    if (employeeId.equals(accepter_user_id)){//被评价人是雇员
                        key = EVALUATE_EE_ORDERID_MAP;
                    }else if (employerId.equals(accepter_user_id)){//被评价人是雇主
                        key = EVALUATE_ER_ORDERID_MAP;
                    }
                    if (StringUtils.isNotEmpty(key)){
                        Redis.hset(key.replace("orderid", orderId), "evaluate_level", evaluate_level);
                    }
                }
            }
        }
    }

    private static void insertOrUpdateUserEnterprise(List<Column> columns) {
        String userId = "";
        for (Column column:columns){
            if ("user_id".equals(column.getName())){
                userId = column.getValue();
                break;
            }
        }
        for (Column column:columns){
            Redis.hset(USER_USERID_ENTERPRESE_EXPAND_MAP.replace("userid",userId),column.getName(), column.getValue());
        }
    }

    private static void insertUserPersonal(List<Column> columns) {
        //所有字段都存入
        String userId = "";
        String credit_count="";
        for (Column column:columns){
            if ("user_id".equals(column.getName())){
                userId = column.getValue();
            }else if ("credit_count".equals(column.getName())){
                credit_count = column.getValue();
            }
        }
        if (StringUtils.isNotEmpty(userId)) {
            for (Column column : columns) {
                Redis.hset(USER_USERID_PERSONAL_EXPAND_MAP.replace("userid", userId), column.getName(), column.getValue());
            }
        }
    }

    private static void updateUserPersonal(List<Column> columns) {
        //所有字段都存入
        String userId = "";
        String credit_count="";
        for (Column column:columns){
            if ("user_id".equals(column.getName())){
                userId = column.getValue();
            }else if ("credit_count".equals(column.getName())){
                credit_count = column.getValue();
            }
        }
        if (StringUtils.isNotEmpty(userId)) {
            for (Column column : columns) {
                Redis.hset(USER_USERID_PERSONAL_EXPAND_MAP.replace("userid", userId), column.getName(), column.getValue());
            }
        }
        Map<String,Object> source = new HashMap<>();
        source.put("creditcount",Double.valueOf(credit_count));
        try {
            ElasticsearchUtil.scrollSearchUpdate(ES_DLG_INDEX, ES_USER_TYPE,"userid", userId, source);
        } catch (Exception e) {
            ES_ERROR.error("修改用户诚信值到ES失败",e);
        }
    }

    private static void insertOrUpdateUser(List<Column> columns) {
        String userId = "";
        String active = "";
        for (Column column:columns) {
            if ("id".equals(column.getName())) {
                userId = column.getValue();
            }else if("active".equals(column.getName())){
                active = column.getValue();
            }
        }
        if ("0".equals(active)){
            deleteUser(columns);
        }else {
            for (Column column : columns) {
                Redis.hset(USER_USERID_MAP.replace("userid",userId),column.getName(), column.getValue());
//                if ("username".equals(column.getName())) {
//                    Redis.hset(USER_USERID_EXPAND_MAP.replace("userid", userId), column.getName(), column.getValue());
//                } else if ("phone".equals(column.getName())) {
//                    Redis.hset(USER_USERID_EXPAND_MAP.replace("userid", userId), column.getName(), column.getValue());
//                }
            }
        }
    }

    private static void insertOrder(List<Column> columns) {
        String orderId = "";
        String employeeId = "";
        String jobId = "";
        String status = "";
        for (Column column : columns) {
            if ("id".equals(column.getName())) {
                orderId = column.getValue();
            } else if ("job_id".equals(column.getName())) {
                jobId = column.getValue();
            } else if ("employee_id".equals(column.getName())) {
                employeeId = column.getValue();
            } else if ("status".equals(column.getName())) {
                status = column.getValue();
            }
        }
        String parent_job_id = Redis.hget(JOB_JOBID_MAP.replace("jobid",jobId),"parent_job_id");
        if (StringUtils.isNotEmpty(parent_job_id)){
            Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("jobid", parent_job_id), orderId + "-" + employeeId);
        }
        for (Column column : columns) {
            Redis.hset(ORDER_ORDERID_MAP.replace("orderid", orderId), column.getName(), column.getValue());
        }
        Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("userid", employeeId), orderId);
        Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("jobid", jobId), orderId + "-" + employeeId);
        String jobOrderKeyStatus = getJobOrderStatus(status);
        if (StringUtils.isNotEmpty(jobOrderKeyStatus)) {
            Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderKeyStatus).replace("jobid", jobId), orderId + "-" + employeeId);
            Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderKeyStatus).replace("jobid", parent_job_id), orderId + "-" + employeeId);
        }
    }

    private static void insertJob(List<Column> columns){
        //任务的扩展信息字段
        List<String> exJobMap = new ArrayList<>();
        exJobMap.add("active");
        exJobMap.add("is_farmers_insurance");
        exJobMap.add("status");

        String userId = "";
        String jobId = "";
        for (Column column:columns){
            if ("id".equals(column.getName())){
                jobId = column.getValue();
            }else if ("user_id".equals(column.getName())){
                userId = column.getValue();
            }else if ("user_id".equals(column.getName())){
                userId = column.getValue();
            }
        }
        for (Column column:columns){
            Redis.hset(JOB_JOBID_MAP.replace("jobid",jobId),column.getName(), column.getValue());
        }
        Redis.lpush(JOB_USERID_STATUS099_LIST.replace("userid",userId),jobId);
    }

    private static void updateOrderTime(List<Column> columns){
        String orderId = "";

        List<String> timeColumnName = getTimeColumns();
        for (Column column:columns) {
            if ("order_id".equals(column.getName())){
                orderId = column.getValue();
            }
        }
        for (Column column:columns) {
            if (timeColumnName.contains(column.getName())) {
                Redis.hset(ORDER_ORDERID_MAP.replace("orderid",orderId), column.getName(), column.getValue());
            }
        }

        String redisOrderKeyStatus = "";
        String isInProgress = Redis.hget(ORDER_ORDERID_MAP.replace("orderid",orderId),"is_in_progress");
        String status = Redis.hget(ORDER_ORDERID_MAP.replace("orderid",orderId),"status");
        String employeeId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid",orderId),"employee_id");
        String jobId = Redis.hget(ORDER_ORDERID_MAP.replace("orderid",orderId),"job_id");
        String jobOrderVal = orderId+"-"+employeeId;

        if (isInProgress.equals("1") && !status.equals("30") && !status.equals("40")){
            redisOrderKeyStatus = "020";
        }else if (status.equals("30")){
            redisOrderKeyStatus = "030";
        }else if (status.equals("40")){
            redisOrderKeyStatus = "040";
        }else if (status.equals("50")){
            redisOrderKeyStatus = "050";
        }

        String jobOrderKeyStatus = getJobOrderStatus(status);

        Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("userid",employeeId),0, orderId);
        Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("userid",employeeId), orderId);
        Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("jobid",jobId),0, jobOrderVal);
        Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("jobid",jobId), jobOrderVal);
        if (StringUtils.isNotEmpty(redisOrderKeyStatus)) {
            if (!redisOrderKeyStatus.equals("050")) {
                Redis.lrem(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderKeyStatus).replace("userid", employeeId), 0, orderId);
                Redis.lpush(ORDER_USERID_STATUS099_LIST.replace("099", redisOrderKeyStatus).replace("userid", employeeId), orderId);
            }
        }
        //最新更新的放前面
        if (StringUtils.isNotEmpty(jobOrderKeyStatus)){
            Redis.lrem(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderKeyStatus).replace("jobid",jobId),0, jobOrderVal);
            Redis.lpush(JOBORDER_JOBID_STATUS099_LIST.replace("099", jobOrderKeyStatus).replace("jobid",jobId), jobOrderVal);
        }
    }

    private static List<String> getTimeColumns(){
        List<String> timeColumnName = new ArrayList<>();
        timeColumnName.add("is_cross_day");
        timeColumnName.add("start_year");
        timeColumnName.add("start_month");
        timeColumnName.add("start_day");
        timeColumnName.add("start_hour");
        timeColumnName.add("start_minute");
        timeColumnName.add("start_second");
        timeColumnName.add("end_year");
        timeColumnName.add("end_month");
        timeColumnName.add("end_day");
        timeColumnName.add("end_hour");
        timeColumnName.add("end_minute");
        timeColumnName.add("end_second");
        return timeColumnName;
    }

    private static void insertOrderTime(List<Column> columns){
        String orderId = "";
        List<String> timeColumnName = getTimeColumns();
        for (Column column:columns) {
            if ("order_id".equals(column.getName())) {
                orderId = column.getValue();
            }
        }
        for (Column column:columns) {
            if (timeColumnName.contains(column.getName())) {
                Redis.hset(ORDER_ORDERID_MAP.replace("orderid",orderId), column.getName(), column.getValue());
            }
        }
    }

    private static void insertJobTime(List<Column> columns){
        String jobId = "";
        List<String> timeColumnName = getTimeColumns();
        //找到任务id
        for (Column column:columns) {
            if ("job_task_id".equals(column.getName())) {
                jobId = column.getValue();
            }
        }
        for (Column column:columns){
            if (timeColumnName.contains(column.getName())) {
                Redis.hset(JOB_JOBID_MAP.replace("jobid",jobId),column.getName(), column.getValue());
            }
        }
    }

    private static void updateJobTime(List<Column> columns){
        String jobId = "";
        List<String> timeColumnName = getTimeColumns();
        //找到任务id
        for (Column column:columns) {
            if ("job_task_id".equals(column.getName())) {
                jobId = column.getValue();
            }
        }
        for (Column column:columns){
            if (timeColumnName.contains(column.getName())) {
                Redis.hset(JOB_JOBID_MAP.replace("jobid",jobId),column.getName(), column.getValue());
            }
        }

        String status = Redis.hget(JOB_JOBID_MAP.replace("jobid",jobId),"status");
        String userId = Redis.hget(JOB_JOBID_MAP.replace("jobid",jobId),"user_id");
        Redis.lrem(JOB_USERID_STATUS099_LIST.replace("userid", userId),0,jobId);
        Redis.lpush(JOB_USERID_STATUS099_LIST.replace("userid", userId), jobId);
        if (status.equals("20") || status.equals("30") || status.equals("40")) {
            Redis.lrem(JOB_USERID_STATUS099_LIST.replace("99", status).replace("userid", userId),0,jobId);
            Redis.lpush(JOB_USERID_STATUS099_LIST.replace("99", status).replace("userid", userId), jobId);
        }
    }

    private static void printColumn(List<Column> columns) {
        for (Column column : columns) {
            LOGGER.info(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }

    private static String getStart(List<Column> columns){
        String startYear="",startMonth="",startDay="",startHour="",startMinutes="",startSecond="";
        for (Column column:columns) {
            if ("start_year".equals(column.getName())) {
                startYear = column.getValue();
            } else if ("start_month".equals(column.getName())) {
                startMonth = column.getValue().length() < 2 ? "0" + column.getValue() : column.getValue();
            } else if ("start_day".equals(column.getName())) {
                startDay = column.getValue().length() < 2 ? "0" + column.getValue() : column.getValue();
            } else if ("start_hour".equals(column.getName())) {
                startHour = column.getValue().length() < 2 ? "0" + column.getValue() : column.getValue();
            } else if ("start_minute".equals(column.getName())) {
                startMinutes = column.getValue().length() < 2 ? "0" + column.getValue() : column.getValue();
            } else if ("start_second".equals(column.getName())) {
                startSecond = column.getValue().length() < 2 ? "0" + column.getValue() : column.getValue();
            }
        }
        return new StringBuilder(startYear).append("-").append(startMonth).append("-").append(startDay).append(" ").append(startHour).append(":").append(startMinutes).append(":").append(startSecond).toString();
    }

    private static String getEnd(List<Column> columns){
        String endYear="",endMonth="",endDay="",endHour="",endMinutes="",endSecond="";
        for (Column column:columns) {
            if ("end_year".equals(column.getName())){
                endYear = column.getValue();
            }else if ("end_month".equals(column.getName())){
                endMonth = StringUtils.isEmpty(column.getValue()) ? null : column.getValue().length()<2?"0"+column.getValue():column.getValue();
            }else if ("end_day".equals(column.getName())){
                endDay = StringUtils.isEmpty(column.getValue()) ? null : column.getValue().length()<2?"0"+column.getValue():column.getValue();
            }else if ("end_hour".equals(column.getName())){
                endHour = StringUtils.isEmpty(column.getValue()) ? "00" : column.getValue().length()<2?"0"+column.getValue():column.getValue();
            }else if ("end_minute".equals(column.getName())){
                endMinutes = StringUtils.isEmpty(column.getValue()) ? "00" : column.getValue().length()<2?"0"+column.getValue():column.getValue();
            }else if ("end_second".equals(column.getName())){
                endSecond = StringUtils.isEmpty(column.getValue()) ? "00" : column.getValue().length()<2?"0"+column.getValue():column.getValue();
            }
        }
        if (endYear!=null && endMonth!=null && endDay!=null) {
            return new StringBuilder(endYear).append("-").append(endMonth).append("-").append(endDay).append(" ").append(endHour).append(":").append(endMinutes).append(":").append(endSecond).toString();
        }else{
            return "";
        }
    }
}